package com.dbingtech.iot.mqtt;

import com.dbingtech.iot.eventbus.SensorUpEvent;
import org.apache.commons.lang3.RandomStringUtils;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 订阅MQTT消息，收到后做2件事情：
 *
 * 1. 阀值判断，如果有异常，则需要告警（短信、电话等方式）
 * 2. 写入数据库，为了提高写入效率，并不是来一条写一条数据库，而是批量写入（每次5000条写入）
 */
public class SensorDataConsumer {

    /**
     *  消息重传 (Message Retransmission) 是属于 MQTT 协议标准规范的一部分。协议中规定了作为通信的双方 服务端 和 客户端 对于自己发送到对端的 PUBLISH 消息都应满足其
     *  服务质量 (Quality of Service levels) 的要求。如：
     *
     *  最多一次，意味着可能丢，要看你的场景，比如物联网领域，一些传感器数据要上报，丢掉一些无所谓的，所以可以采取QOS_ALMOST_ONCE，丢掉无所谓的
     */
    public static final int QOS_ALMOST_ONCE = 0;

    /**
     * QoS 1：表示 消息至少送达一次 (At least once delivery)；即发送端会一直重发该消息，除非收到了对端对该消息的确认。意思是在 MQTT 协议的上层（即业务的应用层）相同的 QoS 1 消息可能会收到多次。
     *
     * 最少一次，意味着可能重复
     */
    public static final int QOS_ATLEAST_ONCE = 1;

    /**
     * 表示 消息只送达一次 (Exactly once delivery)；即该消息在上层仅会接收到一次，不多不少，正好一次，这个和flink里面的语义也是差不多的，都是这个意思。
     *
     * 虽然，QoS 1 和 QoS 2 的 PUBLISH 报文在 MQTT 协议栈这一层都会发生重传，但请你谨记的是：
     *
     * 1、QoS 1 消息发生重传后，在 MQTT 协议栈上层，也会收到这些重发的 PUBLISH 消息
     * 2、QoS 2 消息无论如何重传，最终在 MQTT 协议栈上层，都只会收到一条 PUBLISH 消息，具体咋实现的，难道MQTT Broker记住了每条消息的ID？那得多少内存或硬盘才放得下这门多消息
     *
     * https://docs.emqx.net/broker/latest/cn/advanced/retransmission.html
     *
     */
    public static final int QOS_EXACTLY_ONCE = 2;

    private String address;
    private int port;
    private String subTopic;
    private String pubTopic;
    private int qos = QOS_ALMOST_ONCE;

    /**
     * 干净会话标志，表示发布者、订阅者客户端程序是否要建立持久会话，如果您需要一个持久会话，这意味着当订阅服务质量（QoS）1或2时，代理将存储该客户端的所有订阅以及所有丢失的消息。
     * 默认情况下，此选项为选中状态或“是”，这意味着代理将启动新会话，不为客户端存储任何内容，还将清除先前的持久会话中的所有信息
     */
    private boolean cleanSession = true;

    /**
     * 必须全局唯一，所以我们这里用随机字符串
     * 客户端标识符是连接到MQTT代理的每个MQTT客户端的标识符。对于给定的经纪人，每个客户应该是唯一的。代理使用它来识别客户端和客户端的当前状态。
     * 默认情况下会自动生成。如果尝试连接具有相同客户端标识符的两个MQTT客户端，则代理将拒绝连接
     */
    private String clientId;

    public SensorDataConsumer(String address, int port, String subTopic, String pubTopic, int qos, String clientId) {
        this.address = address;
        this.port = port;
        this.subTopic = subTopic;
        this.pubTopic = pubTopic;
        this.qos = qos;
        this.clientId = clientId;
    }

    private MqttClient mqttClient;

    public static final class OnMessageCallback implements MqttCallback {

        private static Logger logger = LoggerFactory.getLogger(OnMessageCallback.class);

        @Override
        public void connectionLost(Throwable throwable) {
            System.out.println("连接断开，可以做重连");
        }

        @Override
        public void messageArrived(String topic, MqttMessage message) throws Exception {
            logger.info("收到消息，主题：{}，Qos：{}，消息内容：{}", topic, message.getQos(), new String(message.getPayload()));
            //TODO有几个问题需要调研：
            /*
            1. 当有很多传感器在发消息的时候，这里只启动一个消费者，够不够？如果启动多个，必然会造成重复消息消息，它不像kafka的CG，多个客户端不会重复收到消息
            2. 来不及消费的消息，MQTT服务器是否会缓存到磁盘上
            3. 是否要创建多个topic，不同的客户往不同的topic上发送？比如A客户有1000个传感器，B客户有400个，C客户有1500个，D客户有200个...，是大家都送到同一个topic上呢，还是多创建几个topic，然后
            分开送到不同的topic上，这里感觉应该是分开创建topic会比较好一些，那就意味着topic是动态创建的，也就意味着消费者也要动态创建
             */
        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {
            logger.info("deliveryComplete：" + token.isComplete());
        }
    }

    public void publish(String content) throws MqttException {
        MqttMessage message = new MqttMessage(content.getBytes());
        message.setQos(qos);
        mqttClient.publish(pubTopic, message);
    }

    public void close() throws MqttException {
        mqttClient.disconnect();
        mqttClient.close();
    }

    public void subscribe(int qos) throws MqttException {
        String broker = String.format("tcp://%s:%d", this.address, this.port);
        MemoryPersistence persistence = new MemoryPersistence();
        mqttClient = new MqttClient(broker, this.clientId, persistence);
        MqttConnectOptions connOpts = new MqttConnectOptions();
//        connOpts.setUserName("emqx_test");
//        connOpts.setPassword("emqx_test_password".toCharArray());
        connOpts.setUserName("admin");
        connOpts.setPassword("public".toCharArray());
        connOpts.setCleanSession(true);

        mqttClient.setCallback(new OnMessageCallback());
        mqttClient.connect(connOpts);
        mqttClient.subscribe(this.subTopic, qos);
    }

    public static void main(String[] args) {
        SensorDataConsumer consumer = new SensorDataConsumer("47.111.101.74", 1883, "$queue/t1", "$queue/t1", 2, RandomStringUtils.randomAscii(6));
        try {
            consumer.subscribe(SensorDataConsumer.QOS_ALMOST_ONCE);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }



}
